package com.soft863.tc.common.bean;

import com.soft863.tc.common.api.Column;
import com.soft863.tc.common.api.RowKey;
import com.soft863.tc.common.api.TableRef;
import com.soft863.tc.common.constant.Names;
import com.soft863.tc.common.constant.ValueConstant;
import com.soft863.tc.common.util.DateUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.NamespaceNotFoundException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.*;

public abstract class BaseDao {

    private final ThreadLocal<Connection> connHolder = new ThreadLocal<>();
    private final ThreadLocal<Admin> adminHolder = new ThreadLocal<>();

    protected void start() {
        getConnection();
        getAdmin();
    }

    protected void end() throws IOException {
        Admin admin = getAdmin();
        if (admin != null) {
            admin.close();
            adminHolder.remove();
        }
        Connection connection = getConnection();
        if (connection != null) {
            connection.close();
            connHolder.remove();
        }
    }

    /*
     * 获取连接对象
     * */
    protected synchronized Connection getConnection() {
        Connection connection = connHolder.get();
        try {
            if (connection == null) {
                Configuration conf = HBaseConfiguration.create();
                conf.set(ValueConstant.HBASE_ZOOKEEPER_QUORUM, ValueConstant.ZOOKEEPER_CLIENT);
                connection = ConnectionFactory.createConnection(conf);
                connHolder.set(connection);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return connection;
    }

    /*
     * 获取admin
     * */
    protected synchronized Admin getAdmin() {
        Admin admin = adminHolder.get();
        try {
            if (admin == null) {
                admin = getConnection().getAdmin();
                adminHolder.set(admin);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return admin;
    }

    /*
     * 创建命名空间，若存在，不创建
     * */
    protected void creatNamespaceNX(String namespace) throws IOException {
        Admin admin = getAdmin();
        try {
            admin.getNamespaceDescriptor(namespace);
        } catch (NamespaceNotFoundException e) {
            //命名空间不存在
            NamespaceDescriptor namespaceDescriptor =
                    NamespaceDescriptor.create(namespace).build();
            admin.createNamespace(namespaceDescriptor);
        }
    }

    /*
     * 创建表，如果表存在删除表 默认一个分区
     * */
    protected void createTableXX(String name, String... families) throws IOException {
        createTableXX(name, null, null, families);
    }

    /*
     * 创建表，如果表存在删除表 自定义分区个数
     * */
    protected void createTableXX(String name, String coProcessorClass, Integer regionCount, String... families) throws IOException {
        Admin admin = getAdmin();
        TableName tableName = TableName.valueOf(name);
        if (admin.tableExists(tableName)) {
            //表存在，删除表
            deleteTable(name);
        }
        createTable(name, coProcessorClass, regionCount, families);
    }

    protected void createTable(String name, String coProcessorClass, Integer regionCount, String... families) throws IOException {
        Admin admin = getAdmin();
        TableName tableName = TableName.valueOf(name);
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);
        if (!"".equals(coProcessorClass) && coProcessorClass != null) {
            tableDescriptorBuilder.setCoprocessor(coProcessorClass);
        }
        if (families == null || families.length == 0) {
            families = new String[1];
            families[0] = Names.CF_DEFAULT.getValue();
        }
        for (String family : families) {
            ColumnFamilyDescriptor columnFamilyDescriptor =
                    ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family)).build();
            tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
        }
        //增加预分区
        //分区键
        if (regionCount == null || regionCount <= 1) {
            admin.createTable(tableDescriptorBuilder.build());
        } else {
            byte[][] splitKey = genSplitKeys(regionCount);
            admin.createTable(tableDescriptorBuilder.build(), splitKey);
        }


    }

    protected void deleteTable(String name) throws IOException {
        Admin admin = getAdmin();
        TableName tableName = TableName.valueOf(name);
        admin.disableTable(tableName);
        admin.deleteTable(tableName);
    }

    protected void putData(String name, Put put) throws IOException {
        //获取表对象
        Connection conn = getConnection();
        Table table = conn.getTable(TableName.valueOf(name));
        //增加数据
        table.put(put);
        //关闭表
        table.close();
    }

    protected void putData(String name, List<Put> puts) throws IOException {
        //获取表对象
        Connection conn = getConnection();
        Table table = conn.getTable(TableName.valueOf(name));
        //增加数据
        table.put(puts);
        //关闭表
        table.close();
    }

    /**
     * 封装逻辑，直接将对象写入HBase中
     *
     * @param obj 对象
     * @throws IOException            '
     * @throws IllegalAccessException '
     */
    protected void putData(Object obj) throws IOException, IllegalAccessException {
        //反射 todo:NB !!!
        Class<?> clazz = obj.getClass();
        //获取TableRef注解的value 即为表名
        TableRef tableRef = clazz.getAnnotation(TableRef.class);
        //表名
        String name = tableRef.value();
        //获取所有属性
        Field[] fields = clazz.getDeclaredFields();

        //rowKey
        String StringRowKey = "";
        for (Field field : fields) {
            //获取被RowKey注解的属性
            RowKey rowKey = field.getAnnotation(RowKey.class);
            if (rowKey != null) {
                field.setAccessible(true);
                //获取属性的值
                StringRowKey = (String) field.get(obj);
                break;
            }
        }
        Put put = new Put(Bytes.toBytes(StringRowKey));

        //获取所有列族 列名和数据
        for (Field field : fields) {
            Column column = field.getAnnotation(Column.class);
            if (column != null) {
                String family = column.family();
                String colName = column.column();
                if ("".equals(colName)) {
                    colName = field.getName();
                }

                field.setAccessible(true);
                String value = (String) field.get(obj);

                put.addColumn(Bytes.toBytes(family), Bytes.toBytes(colName), Bytes.toBytes(value));
            }
        }

        //获取表对象
        Connection conn = getConnection();
        Table table = conn.getTable(TableName.valueOf(name));
        //增加数据
        table.put(put);
        //关闭表
        table.close();
    }

    /**
     * 获取查询时startRow、StopRow集合
     * 如：查询133 在2019-09 ~ 2020-09 所有通话记录
     * 因为分区号是散列生成的，因此只能一个月一个月的查询，即
     * x_133_2019_09 ~ x_133_2019_09|
     * x_133_2019_10 ~ x_133_2019_10|
     * ...
     * x_133_2020_01 ~ x_133_2020_09|
     *
     * @param tel   电话
     * @param start 开始时间
     * @param end   结束时间
     * @return tel 在 start - end 之间通话的查询条件
     */
    protected List<String[]> getStartStopRowKeys(String tel, String start, String end) {
        List<String[]> rowKeys = new ArrayList<>();
        Calendar calendar = Calendar.getInstance();
        String startTime = start.substring(0, 6);
        String endTime = end.substring(0, 6);
        String format = "yyyyMM";
        Date startDate = DateUtil.parse(startTime, format);
        Date endDate = DateUtil.parse(endTime, format);

        int regionNum = 0;
        String startStopRowKey = "";
        do {
            startStopRowKey = DateUtil.format(startDate.getTime(), format);
            regionNum = genRegionNum(tel, startStopRowKey);
            String startRow = regionNum + "_" + tel + "_" + startStopRowKey;
            String stopRow = startRow + "|";
            String[] tmpRowKeys = {startRow, stopRow};
            rowKeys.add(tmpRowKeys);
            calendar.setTime(startDate);
            calendar.add(Calendar.MONTH, 1);
            startDate = calendar.getTime();
        } while (startDate.compareTo(endDate) <= 0);

        return rowKeys;
    }

    private byte[][] genSplitKeys(int regionCount) {
        int splitKeyCount = regionCount - 1;
        byte[][] bs = new byte[splitKeyCount][];
        // 00011     11000     22333
        // | 的ascii码第二大
        // (-∞, 0|),[0|, 1|),[1|, +∞)
        ArrayList<byte[]> bsList = new ArrayList<>();
        for (int i = 0; i < splitKeyCount; i++) {
            String splitKey = i + "|";
            bsList.add(Bytes.toBytes(splitKey));
        }
        //排序
        bsList.sort(new Bytes.ByteArrayComparator());
        return bsList.toArray(bs);
    }

    /*
     * 根据业务计算分区号，保证同一个号码在同一个年同一月在一个分区
     * */
    protected int genRegionNum(String tel, String date) {
        String userCode = tel.substring(tel.length() - 4);
        String yearMonth = date.substring(0, 6);

        int userCodeHash = userCode.hashCode();
        int yearMonthHash = yearMonth.hashCode();
        //异或算法 crc校验
        int crc = Math.abs(userCodeHash ^ yearMonthHash);
        //取模，位运算有限制
        return crc % ValueConstant.REGION_COUNT;
    }

}
